// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu, gekim

#pragma once

#include <algorithm>
#include <memory>
#include <set>
#include <stdexcept>
#include <string>
#include <utility>
#include <vector>

#include <fmt/format.h>

#include "coin/feed/fastfeed/feed.pb.h"
#include "feedset/presto/common/proto/archive_feed.pb.h"
#include "feedset/presto/common/proto/enums.pb.h"
#include "feedset/presto/common/proto/feed.pb.h"
#include "orio/io/base.h"
#include "orio/record/simple.h"

#include "coin2/base/time.h"
#include "coin2/base/log.h"
#include "coin2/feed/fastfeed/reader.h"

using NaviFeed = presto::feed_subscriber::proto_feed::ArchiveFeedMessage;
using CoinFeed = fastfeed::proto::Feed;

// this class reads only one stream
class NaviFeedProtoReader : public orio::io::Reader<NaviFeed> {
 public:
  NaviFeedProtoReader(
      std::unique_ptr<orio::io::Reader<orio::record::SimpleRecord>> reader,
      size_t max_record_size)
      : reader_(std::move(reader)), buf_(new char[max_record_size]) {}

  bool Read(NaviFeed* out) override {
    if (!reader_->Read(
            reinterpret_cast<orio::record::SimpleRecord*>(buf_.get()))) {
      return false;
    }
    auto* r = reinterpret_cast<orio::record::SimpleRecord*>(buf_.get());
    if (!out->ParseFromArray(r->data, r->len)) {
      throw std::runtime_error("[NaviFeedProtoReader] fails to parse proto");
    }
    return true;
  }

 private:
  std::unique_ptr<orio::io::Reader<orio::record::SimpleRecord>> reader_;
  std::unique_ptr<char[]> buf_;
};

class NaviPerSymbolReader : public orio::io::Reader<NaviFeed> {
 public:
  explicit NaviPerSymbolReader(std::unique_ptr<NaviFeedProtoReader> reader)
      : reader_(std::move(reader)) {
    header_.Clear();
  }

  bool Read(NaviFeed* out) override;

 private:
  std::string symbol_;
  presto::feed_subscriber::proto_feed::ArchiveFeedMessage_DuplicatedHeader
      header_;
  std::unique_ptr<NaviFeedProtoReader> reader_{nullptr};
};

class NaviToFastfeedConverter : public orio::io::Reader<CoinFeed> {
 public:
  explicit NaviToFastfeedConverter(std::unique_ptr<orio::io::Reader<NaviFeed>> reader)
      : reader_(std::move(reader)) {}

  bool Read(CoinFeed* out) override {
    while (true) {
      if (!reader_->Read(&buf_)) {
        return false;
      }

      if (buf_.feed_type() != presto::FeedType::BOOK_FEED &&
          buf_.feed_type() != presto::FeedType::TRADE_FEED) {
        continue;
      }

      break;
    }

    // only process BOOK or TRADE

    out->clear_custom_feed_type();                                                 // 2
    out->set_feed_continue(fastfeed::proto::FeedContinue::FEED_CONTINUE_UNKNOWN);  // 3

    out->set_symbol(buf_.duplicated_header().symbol());  // 5
    out->set_timestamp(buf_.fetched_time());             // 6
    out->set_origin_timestamp(buf_.exchange_time());     // 7
    out->set_sort_key(0);                                // 8

    out->clear_feed_arb_param();      // 10
    out->clear_book_builder_param();  // 11
    out->clear_trade();               // 20
    out->clear_book_snapshot();       // 21
    out->clear_book_update();         // 22
    out->clear_liquidation();         // 23
    out->clear_string();              // 30
    out->clear_json();                // 31
    out->clear_json_object_table();   // 32

    if (buf_.feed_type() == presto::FeedType::BOOK_FEED) {
      out->set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_BOOK);           // 1
      out->set_data_type(fastfeed::proto::DataType::DATA_TYPE_BOOK_SNAPSHOT);  // 4

      // diff is hard to handle
      auto& coin_book = *out->mutable_book_snapshot();
      for (const auto& entry : buf_.ask_book()) {
        auto& coin_entry = *coin_book.add_ask();
        coin_entry.set_price(entry.price());
        coin_entry.set_qty(entry.qty());
        coin_entry.set_count(entry.count());
      }
      for (const auto& entry : buf_.bid_book()) {
        auto& coin_entry = *coin_book.add_bid();
        coin_entry.set_price(entry.price());
        coin_entry.set_qty(entry.qty());
        coin_entry.set_count(entry.count());
      }

      // // ask

      // if (buf_.ask_type() == presto::feed_subscriber::proto_feed::BookType;:NORMAL_BOOK) {
      //   ask clear;
      //   const auto& navi_book = buf_.ask_book();
      // } else if (buf_.ask_type() == presto::feed_subscriber::proto_feed::BookType;:DIFF_BOOK) {
      //   for(const auto& diff_enrry: buf_.ask_book_diff()) {

      //   }
      // }

      // // bid

      // ;

      // if (buf_.has_book()) {
      //   out->set_data_type(fastfeed::proto::DataType::DATA_TYPE_BOOK_SNAPSHOT);  // 4

      //   auto& coin_book = *out->mutable_book_snapshot();

      //   for (const auto& entry : navi_book.ask()) {
      //     auto& coin_entry = *coin_book.add_ask();
      //     coin_entry.set_price(entry.price());
      //     coin_entry.set_qty(entry.qty());
      //     coin_entry.set_count(entry.count());
      //   }
      //   for (const auto& entry : navi_book.bid()) {
      //     auto& coin_entry = *coin_book.add_bid();
      //     coin_entry.set_price(entry.price());
      //     coin_entry.set_qty(entry.qty());
      //     coin_entry.set_count(entry.count());
      //   }
      // } else {
      //   throw std::runtime_error("we expect navi has no delta feed");

      //   out->set_data_type(fastfeed::proto::DataType::DATA_TYPE_BOOK_UPDATE);  // 4

      //   const auto& navi_book = buf_.book_diff();
      //   auto& coin_book = *out->mutable_book_snapshot();

      //   for (const auto& diff_entry : navi_book.ask()) {
      //     const auto& entry = diff_entry.book_entry();
      //     auto& coin_entry = *coin_book.add_ask();
      //     coin_entry.set_price(entry.price());
      //     coin_entry.set_qty(entry.qty());
      //     coin_entry.set_count(entry.count());
      //   }
      //   for (const auto& diff_entry : navi_book.bid()) {
      //     const auto& entry = diff_entry.book_entry();
      //     auto& coin_entry = *coin_book.add_bid();
      //     coin_entry.set_price(entry.price());
      //     coin_entry.set_qty(entry.qty());
      //     coin_entry.set_count(entry.count());
      //   }
      // }

    } else {
      out->set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADE);  // 1
      out->set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADE);  // 4

      const auto& navi_trade = buf_.trade();
      auto& coin_trade = *out->mutable_trade();

      coin_trade.set_price(navi_trade.price());
      coin_trade.set_qty(navi_trade.qty());

      auto&& side = navi_trade.side();
      if (side == presto::TradeSide::TRADE_NO_SIDE) {
        coin_trade.set_side(fastfeed::proto::TradeSide::TRADE_SIDE_INVALID);
      } else if (side == presto::TradeSide::TRADE_BUY_SIDE) {
        coin_trade.set_side(fastfeed::proto::TradeSide::TRADE_SIDE_BUY);
      } else if (side == presto::TradeSide::TRADE_SELL_SIDE) {
        coin_trade.set_side(fastfeed::proto::TradeSide::TRADE_SIDE_SELL);
      } else {
        throw std::runtime_error("Unknown navi's TradeSide. Please update this code.");
      }
    }

    return true;
  }

 private:
  std::unique_ptr<orio::io::Reader<NaviFeed>> reader_;
  NaviFeed buf_{};
};

// this class is responsible for reading multiple files & merge
class NaviUniverseFeedReader : public orio::io::Reader<CoinFeed> {
 public:
  // /remote/iosg/data/repository/feed/universe/{machine}/{universe}/{YYMMDD}/a
  // /remote/iosg/data/repository/feed/universe/{machine}/{universe}/{YYMMDD}/b
  // /remote/iosg/data/repository/feed/universe/{machine}/{universe}/{YYMMDD}/c
  //
  // /remote/iosg/data/repository/feed/proto/{machine}/{exchange}/{YYYYMMDD}/{product}/0.day.20200508-153739+0900
  NaviUniverseFeedReader(
      const std::experimental::filesystem::path& universe_root_dir,
      const std::string& machine,
      const std::string& universe,
      const std::set<std::string>& channels,
      int64_t warmup,
      int64_t begin_timestamp, /* inclusive */
      int64_t end_timestamp,   /* exclusive */
      size_t max_record_size) {
    using namespace fmt::literals;
    std::vector<std::unique_ptr<orio::io::Reader<CoinFeed>>> readers;

    std::vector<std::experimental::filesystem::path> paths = GetSortedPathList(
        universe_root_dir,
        machine,
        universe,
        channels,
        begin_timestamp - warmup,
        end_timestamp);

    for (auto& path : paths) {
      LOG(INFO) << "[NaviUniverseFeedReader] will read " << path;
    }

    reader_ = std::unique_ptr<orio::io::Reader<CoinFeed>>(new IntervalFastFeedReader(
        std::unique_ptr<orio::io::Reader<CoinFeed>>(new NaviToFastfeedConverter(
            std::unique_ptr<orio::io::Reader<NaviFeed>>(new NaviFeedProtoReader(
                std::unique_ptr<orio::record::SimpleMultiFileReader>(
                    orio::record::SimpleMultiFileReader::FromFile(paths, max_record_size)),
                max_record_size)))),
        begin_timestamp - warmup,
        end_timestamp));
  }

  bool Read(CoinFeed* out) override { return reader_->Read(out); }

  static std::vector<std::experimental::filesystem::path> GetSortedPathList(
      const std::experimental::filesystem::path& universe_root_dir,
      const std::string& machine,
      const std::string& universe,
      const std::set<std::string>& channels,
      int64_t begin_timestamp,
      int64_t end_timestamp /* exclusive */) {
    std::vector<std::pair<std::string, std::experimental::filesystem::path>> paths;

    auto p = PtimeFromTimestamp(begin_timestamp);
    begin_timestamp = TimestampFromPtime(
        PtimeFromDatetime(p.date().year(), p.date().month(), p.date().day(), 0, 0, 0, 0));

    for (int64_t ts = begin_timestamp; ts < end_timestamp; ts += 24 * 60 * 60 * 1'000'000'000LL) {
      auto yyyymmdd = YmdFromTimestamp(ts);
      auto dir = universe_root_dir / machine / universe / yyyymmdd;
      if (!std::experimental::filesystem::exists(dir)) {
        LOG(WARNING) << "Missing date from stock universe feed: " << dir;
        continue;
      }

      for (auto& p : std::experimental::filesystem::directory_iterator(dir)) {
        /// "--" => recipe
        // no "--" => empty recipe
        // 1.morning.20200715-093000+0800--tag..

        if (std::experimental::filesystem::is_directory(p)) {
          continue;
        }
        auto filename = p.path().filename().string();
        auto idx = filename.find("--");
        std::string_view session;
        std::string tag;
        if (idx == std::string::npos) {
          session = filename;
          tag = "";
        } else {
          session = filename.substr(0, idx);
          tag = filename.substr(idx + 2);
        }
        if (channels.find(tag) != channels.end()) {
          CHECK_THROW(session.size() >= 22);  // 22 = dates + two dots
          std::string datetime(session.substr(session.size() - 20, 20));
          paths.push_back(std::make_pair(datetime, p));
        }
      }
    }

    std::sort(paths.begin(), paths.end());
    std::vector<std::experimental::filesystem::path> out;

    for (const auto& pair : paths) {
      out.push_back(pair.second);
    }

    return out;
  }

 private:
  std::unique_ptr<orio::io::Reader<CoinFeed>> reader_;
};

class NaviPerSymbolFeedReader : public orio::io::Reader<CoinFeed> {
 public:
  // /remote/iosg/data/repository/feed/proto/{machine}/{exchange}/{YYYYMMDD}/{product}/0.day.20200508-153739+0900
  NaviPerSymbolFeedReader(
      const std::experimental::filesystem::path& per_symbol_root_dir,
      const std::string& machine, const std::string exchange_str,
      const std::set<std::string>& symbols, int64_t warmup,
      int64_t begin_timestamp, /* inclusive */
      int64_t end_timestamp,   /* exclusive */
      size_t max_record_size);

  static std::vector<std::experimental::filesystem::path> GetSortedPathList(
      const std::experimental::filesystem::path& per_symbol_root_dir,
      const std::string& machine, const std::string& exchange_str,
      const std::set<std::string>& symbols, int64_t begin_timestamp,
      int64_t end_timestamp /* exclusive */);

  bool Read(CoinFeed* out);

 private:
  std::unique_ptr<orio::io::Reader<CoinFeed>> reader_{nullptr};
  std::vector<std::unique_ptr<char[]>> buffers_;
};
